package com.cloud.core.graphx.demo

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Graph_learning_02 {

  var sc: SparkContext = null
  val master = "local"
  val appName = "Graph_learning_02"

  // 初始化SparkContext
  def init(): Unit = {

    val sparkConf = new SparkConf()
      .setMaster(master)
      .setAppName(appName)

    sc = new SparkContext(sparkConf)
  }

  def main(args: Array[String]): Unit = {

    init()

    // Create an RDD for the vertices
    val users: RDD[(VertexId, (String, String))] =
      sc.parallelize(Seq((3L, ("rxin", "student")),
        (7L, ("jgonzal", "postdoc")),
        (5L, ("franklin", "prof")),
        (2L, ("istoica", "prof")),
        (4L, ("peter", "student"))))
    // Create an RDD for edges
    val relationships: RDD[Edge[String]] =
      sc.parallelize(Seq(Edge(3L, 7L, "collab"),
        Edge(5L, 3L, "advisor"),
        Edge(2L, 5L, "colleague"),
        Edge(5L, 7L, "pi"),
        Edge(4L, 0L, "student"),
        Edge(5L, 0L, "colleague")))

    // Define a default user in case there are relationship with missing user
    val defaultUser = ("John Doe", "Missing")
    // Build the initial Graph
    val graph = Graph(users, relationships, defaultUser)
    // Notice that there is a user 0 (for which we have no information) connected to users
    // 4 (peter) and 5 (franklin).
    graph.triplets.map(
      triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
    ).collect.foreach(println(_))

    println("-----------------------------")

    val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing")

    validGraph.triplets.map(
      triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
    ).collect.foreach(println(_))


    val outDegrees = validGraph.outDegrees

    val outDegGraph = graph.outerJoinVertices(outDegrees) {
      (id, oldAttr, outDegOpt) =>
      outDegOpt match {
        case Some(outDeg) => outDeg
        case None => 0 // No outDegree means zero outDegree
      }
    }

  outDegGraph.vertices.map(v => v._1 + "," + v._2)
    .collect().foreach(println(_))

}
}
